package org.raymond.iworks.study.mq.kafka.demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.Set;

/**
 * 消费组概念
 * group id相同就属于同一个消费组
 * 1）每个consumer都要属于一个consumer.group,就是一个消费组,topic的一个分区只会分配给 一个消费组下的一个consumer来处理,
 * 每个consumer可能会分配多个分区,也有可能某个consumer没有分配到任何分区
 * 2）如果想要实现一个广播的效果,那只需要使用不同的group id去消费就可以. 事例如下:
 * 设topicA包含 partition0和partition1 两个分区
 * groupA：consumer1:分配到 partition0, consuemr2:分配到 partition1, consuemr3: 无分区分配.
 * groupB: consuemr3:分配到 partition0和partition1
 * 3）如果consumer group中某个消费者挂了,此时会自动把分配给他的分区交给其他的消费者,如果他又重启了,那么又会把一些分区重新交还给他
 *
 * offset是以groupid-topic-partition为key保存到broker的内部主题中的.因此不同的groupid,可以接收到相同的message,
 * 但同topic下,同一个groupid同一个partition,只能有一个消费者能消费到message
 *
 * new KafkaConsumer对象后,在while true循环中执行consumer.poll操作拉取消息中,会有两个线程执行:一个是heartbeat 线程,
 * 另一个是processing线程
 *  * processing 线程可理解为调用consumer.poll方法执行消息处理逻辑的线程.
 *  * heartbeat 线程是一个后台线程,对程序员是"隐藏不见"的.heartbeat 线程每隔heartbeat.interval.ms向coordinator发送一个心跳包,
 *  证明自己还活着.但是如果group coordinator在一个heartbeat.interval.ms周期内未收到consumer的心跳,就把该consumer移出group,
 *  这有点说不过去.事实上,有可能网络延时,有可能consumer出现了一次长时间GC,影响了心跳包的到达,说不定下一个heartbeat就正常了.
 *  而如果group coordinator在session.timeout.ms 内未收到consumer的心跳,那就会把该consumer移出group.
 */
/**
 * 案例:二手电商平台(欢乐送),根据用户消费的金额,对用户星星进行累计.
 * 订单系统(生产者)->Kafka集群里面发送了消息.
 * 会员系统(消费者)->Kafak集群里面消费消息,对消息进行处理
 */
/**
 * 消费者是如何实现rebalance的? 根据coordinator实现.
 * 什么是coordinator? 每个consumer group都会选择一个broker作为自己的coordinator,他是负责监控这个消费组里的各个消费者的心跳,
 * 判断是否有消费者退出(宕机或掉线等原因)或新加入,然后开启rebalance.
 * 如何选择coordinator. 首先对groupId进行hash(数字),接着对__consumer_offsets的分区数量取模,默认数量是50,__consumer_offsets的
 * 分区数可以通过offsets.topic.num.partitions来设置,找到分区以后,这个分区所在的broker机器就是coordinator机器.
 * 比如说：groupId,“myconsumer_group” -> hash值(数字)-> 对50取模 -> 8 __consumer_offsets 这个主题的8号分区在哪台broker上面,
 * 那一台就是coordinator, 就知道这个consumer group下的所有的消费者提交offset的时候是往哪个分区去提交offset.
 * 初始化流程:
 * 1）每个consumer都发送JoinGroup请求到Coordinator.
 * 2）然后Coordinator从一个consumer group中选择一个consumer作为leader.
 * 3）把consumer group要消费的topic的情况发送给这个leader.
 * 4）接着这个leader会负责制定消费方案.
 * 5）通过SyncGroup发给Coordinator.
 * 6）接着Coordinator就把消费方案下发给各个consumer,他们会从指定的分区的leader开始进行socket连接以及消费消息
 * 7) 每个消费者都会和coordinator保持心跳(3s),一旦超时(session.timeout.ms=45s),该消费者将从ISR中移除,并触发再平衡.
 * 或者消费者处理消息的时间太长,超过(max.poll.interval.ms=5min),也会触发再平衡.
 */
/**
 * 消费流程:
 * 1) 创建一个ConsumerNetworkClient.
 * 2) sendFetches: 发送请求到ConsumerNetworkClient,初始化抓取数据相关参数.
 * 准备参数:
 * fetch.min.bytes:每批次至少抓取大小,默认1 byte;
 * fetch.max.wait.ms:等待一批数据达到最小抓取大小的超时时间,默认500ms;
 * fetch.max.bytes:每批次最大抓取大小,默认50 Mb;
 * 3) 调用send方法发送请求到broker
 * completedFetches.onSucess: 拉取数据放入一个queue中.
 * 4) FetchedRecords:从队列中抓取数据
 * max.poll.records 一次拉取数据返回消息的最大条数,默认500条.
 * parseRecord: 反序列化.对应producer的序列化
 * Interceptors: 拦截器
 */
/**
 * 分区分配策略
 * Range: Range是对每个topic而言的.
 *   首先对同一个topic里面的分区按照序号进行排序,并对消费者按照字母顺序进行排序.
 * 假如现在有7个分区,3个消费者,排序后的分区会是0,1,2,3,4,5,6;消费者排序后会是C0,C1,C2.
 * 通过partition数/consumer数来决定每个消费者应该消费几个分区.如果除不尽,那么前面几个消费者将会多消费1个分区.
 * 例如,7/3=2余1,除不尽,那么消费者C0便会多消费1个分区.8/3=2余2,那么C0和C1分别多消费1个.
 * 注意:如果只针对1个topic而言,C0消费者多消费1个分区影响不是很大.但是如果有N个topic,那么针对每个topic,消费者C0
 * 都将多消费1个分区,topic越多,C0消费者会被其他消费者明显多消费N个分区.容易产生数据倾斜!
 * RoundRobin以及再平衡: 针对集群中所有Topic而言.
 *   轮询分区策略,是把所有的partition和所有的consumer都列出来,然后按照hashcode进行排序,最后通过轮询算法来分配
 * partition给各个消费者.
 * Sticky以及再平衡:
 * 粘性分区定义:可以理解为分配的结果带有"粘性".在执行一次新的分配之前,考虑上一次分配的结果,尽量少的调整分配的变动,
 * 可以节省大量的开销.
 * 首先会尽量均衡的放置分区到消费者上面.在出现同一消费者组内消费者出现的问题的时候,会尽量保持原有的分区不变.
 */

/**
 * 消息堆积:
 * 1) 如果是kafka消费能力不足,考虑增加topic分区数,并同时提升消费组的消费者数量,消费者数=分区数(缺一不可)
 * 2) 如果是下游的数据处理不及时:提高每批次拉取的数量. 批次拉取数据过少(拉取数据/处理时间<生产速度),使处理的数据小于生产的数据,也会造成数据积压.
 */
@Slf4j
public class ConsumerTest {
    private static final String kafkaAddress = "192.168.0.125:9094,192.168.0.125:9194,192.168.0.125:9294";

    public static void main(String[] args) throws Exception {
        consumer1();
    }

    public static void consumer1(){
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaAddress);
        // 指定消费者组id
        // 消费者组内不同的消费者负责消费不同分区的消息,一个分区只能由组内一个消费者消费.
        // 消费者组之间互不影响.所有消费者都来自于某个消费者组,即消费者组是一个逻辑订阅者.
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        // 序列化
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        /**
         * 每个consumer 都会根据heartbeat.interval.ms参数指定的时间周期性地向group coordinator发送hearbeat,
         * group coordinator会给各个consumer响应.若发生了rebalance,各个consumer收到的响应中会包含REBALANCE_IN_PROGRESS标识,
         * 这样各个consumer就知道已经发生了rebalance,同时group coordinator也知道了各个consumer的存活情况.
         */
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 10000);
        /**
         * group coordinator检测consumer失效所需的时间.一个consumer group里面的某个consumer挂掉了,
         * 最长需要session.timeout.ms毫秒检测出来.它指定了一个阈值—10秒,在这个阈值内如果group coordinator未收到
         * consumer的任何消息(指心跳),那coordinator就认为consumer挂了.
         */
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 20000);
        /**
         * 如果consumer两次poll操作间隔超过了这个时间,broker就会认为这个consumer处理能力太弱,会将其踢出消费组,
         * 将分区分配给别的consumer消费,并触发rebalance.
         */
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 15000);
        /**
         * 最小可获取消息的字节数,默认1字节
         */
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);
        /**
         * 最大可获取消息的字节数,默认为50M.
         */
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 5*1024*1024);
        /**
         * 等待一批数据达到最小抓取大小的超时时间,默认500ms
         */
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 5000); // default: 500
        /**
         * 一次poll返回消息的最大条数,默认是500条
         */
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        /**
         * consumer跟broker的socket连接如果空闲超过了一定的时间,此时就会自动回收连接,但是下次消费就要重新建立socket连接.
         * 这个建议设置为-1，不回收
         */
        props.put(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, -1);
        /**
         * 是否开启自动提交偏移量
         */
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        /**
         * 开启自动提交后,每隔多久提交一次偏移量,默认值5000毫秒
         */
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
        /**
         * earliest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费:
         * topica -> partition0:1000 partitino1:2000
         * latest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,自动将偏移量重置为最新的偏移量.
         * none: topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常.
         */
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 消费指定主题
        String topic = "my-topic";
        consumer.subscribe(Arrays.asList(topic));

        // 消费指定分区
        /*
        List<TopicPartition> topicPartitions = new ArrayList<>();
        topicPartitions.add(new TopicPartition(topic, 0));

        consumer.assign(topicPartitions);
         */

        Set<TopicPartition> partitions = consumer.assignment();
        // 需要保证获取到分区分配方案
        while(partitions.size()==0){
            consumer.poll(Duration.ofSeconds(1));
            partitions = consumer.assignment();
        }
        /*
        // 以时间追溯offset
        Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
        for(TopicPartition partition : partitions){
            // 查找1小时前的offset的位置
            timestampsToSearch.put(partition, System.currentTimeMillis() - 3600 * 1000);
        }
        Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestampMap = consumer.offsetsForTimes(timestampsToSearch);

        for(TopicPartition partition : partitions){
            //consumer.seek(partition, 3);

            OffsetAndTimestamp offsetAndTimestamp = offsetAndTimestampMap.get(partition);
            if(offsetAndTimestamp!=null) {
                consumer.seek(partition, offsetAndTimestamp.offset());
            }else{
                consumer.seek(partition, 33);
            }
        }*/

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));
            for (ConsumerRecord<String, String> record : records) {
                //System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                log.info("recv record: {}", record);
            }
            consumer.commitSync();
        }
    }
}
